package xdu.lz.stage6_threadpool.chapter3_threadPoolExecutorAPI;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;

/**
 * Created at 2019/10/24 0024 下午 3:39
 * <p>
 * 批量提交任务，获取结果是阻塞的
 * invoke all(List<Runnable> task) --> blocked until all the threads finish it's work
 * invoke any --> blocked until any thread finish it's work
 * <p>
 * <p>
 * 缺点：批量提交任务获取结果的时候，运气不好获取的是最长时间的那个，会导致最长时间完成才能获取到其他的。
 * <p>
 * 比如：要采集100条数据，开启了200个线程，只要前100个采集完毕后，就把计划给取消。
 * 运气不好的话 ，getResult的时候可能拿的是最耗时间的那个。
 * <p>
 * JDK1.8之前用了：
 * 见{@link ExecutorCompletionService}，内置了一个BlockingQueue，会把先执行完的结果放到queue中，
 * 测试见{@link ExecutorCompletionServiceDemo}
 *
 * JDK1.8之后用了
 * 见{@link CompletableFuture}
 * 测试见{@link CompletableFuture1}
 */
public class InvokeAny_All_Callable {

    public static void main(String[] args) {
        invokeAny();
        invokeAll();
    }

    /**
     * 测试：提交任务阻塞住，直到任意一个完成后返回并取消其他任务
     * {@link ExecutorService#invokeAny(Collection)}
     * {@link ExecutorService#invokeAny(Collection, long, TimeUnit)}
     */
    private static void invokeAny() {
        ExecutorService executor = Executors.newCachedThreadPool();
        List<Callable<Integer>> taskList = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            int finalI = i;
            Callable<Integer> callable = new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {
                    Thread.sleep(new Random().nextInt(2000) * 1000);
                    System.out.println(Thread.currentThread().getName() + "" + finalI);
                    return finalI;
                }
            };
            taskList.add(callable);
        }
        //blocked until any thread finish it's work ,other tasks will be canceled
        // if timeOut,cancled all
        try {
            Integer any = executor.invokeAny(taskList);
            Integer any1 = executor.invokeAny(taskList, 100, TimeUnit.SECONDS);
            System.out.println("any = " + any);
            System.out.println("any1 = " + any1);
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("finished ================= ");
    }

    /**
     * 测试：提交任务后阻塞住，直到所有任务完成
     * {@link ExecutorService#invokeAll(Collection)} }
     * {@link ExecutorService#invokeAll(Collection, long, TimeUnit)}
     */
    private static void invokeAll() {
        ExecutorService executor = Executors.newCachedThreadPool();
        List<Callable<Integer>> taskList = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            int finalI = i;
            Callable<Integer> callable = new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {
                    Thread.sleep(new Random().nextInt(2000) * 1000);
                    System.out.println(Thread.currentThread().getName() + "" + finalI);
                    return finalI;
                }
            };
            taskList.add(callable);
        }

        //blocked until all the threads finish it's work
        // if timeOut.give up and cancled all tasks
        try {
            List<Future<Integer>> futures = executor.invokeAll(taskList);
            List<Future<Integer>> futures1 = executor.invokeAll(taskList, 10, TimeUnit.MILLISECONDS);
            for (Future<Integer> future : futures) {
                System.out.println(future.get());
            }
            for (Future<Integer> future : futures1) {
                System.out.println(future.get());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("finished ================= ");

    }
}
